{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "chapter-intro",
   "metadata": {},
   "source": [
    "# 第 4 章：AI 智能体的交互体验\n",
    "\n",
    "> 本笔记文件需要与《LangGraph实战》的第 4 章的内容配套使用。\n",
    "\n",
    "在 AI 应用快速发展的今天，用户体验已成为应用成败的关键。用户不仅期望 AI 智能体具备高度智能，更追求流畅、快速、自然的交互体验，以及智能体在长期互动中展现出的记忆力和理解力。\n",
    "\n",
    "本章将深入探讨 LangGraph 框架如何通过流式处理、持久化和人机环路协作三大核心机制，全面提升 AI 智能体的用户体验，打造更值得信赖、更具吸引力、更高效实用的 AI 系统。\n",
    "\n",
    "我们将揭示 LangGraph 如何利用流式处理技术，克服 LLM 固有的延迟，实现智能体输出的实时反馈，提升交互的即时性和流畅感。我们将深入探索 LangGraph 的持久化机制，阐释其如何赋予智能体\"记忆力\"，实现跨会话状态保持、断点续传、时间旅行调试等高级功能。我们还将重点介绍如何通过人机环路协作，实现人工用户对智能体行为的有效指导和监督，构建更符合人类价值观、更值得信赖的 AI 系统。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0omlhyyoq2c",
   "metadata": {},
   "source": [
    "### 🚀 环境准备\n",
    "\n",
    "首先加载必要的环境变量配置："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "6d72b951",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "True"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from dotenv import load_dotenv\n",
    "\n",
    "load_dotenv()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "z3qgqe21voc",
   "metadata": {},
   "source": [
    "## 4.1 实时互动：掌握 LangGraph 中的流式处理\n",
    "\n",
    "在人工智能智能体领域，用户体验 (User Experience，UX) 至关重要。无论一个智能智能体的推理能力多么复杂，功能多么强大，如果用户感觉它速度缓慢或反应迟钝，都将无法达到用户的期望。\n",
    "\n",
    "流式处理作为一种强大的技术应运而生，可以克服延迟挑战，并显著提高使用 LangGraph 构建的 AI 智能体的感知响应速度。流式处理不是等待整个输出生成完毕后再显示给用户，而是允许以分块的方式逐步交付信息。\n",
    "\n",
    "LangGraph 在架构设计之初就优先考虑了对流式处理的一流支持。它提供了一套灵活的流式处理模式，每种模式都针对不同的用例以及监控和用户反馈所需的详细程度量身定制。\n",
    "\n",
    "### 4.1.1 探索流式处理模式：根据用户需求定制输出\n",
    "\n",
    "LangGraph 通过 `.stream()` 和 `.astream()` 方法提供多样化的流式处理模式，使开发人员能够选择在图执行期间流式传输的数据类型和粒度。`stream_mode` 参数是解锁这些功能的关键。\""
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7xi8ro7xnbd",
   "metadata": {},
   "source": [
    "##### 示例 4-1：一个简单的笑话生成图\n",
    "\n",
    "让我们首先创建一个简单的 LangGraph，它旨在优化用户提供的主题，然后围绕该主题生成一个笑话。这将帮助我们理解不同的流式处理模式。\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "lay7wu1msd",
   "metadata": {},
   "outputs": [],
   "source": [
    "from typing import TypedDict\n",
    "from langgraph.graph import StateGraph, START\n",
    "\n",
    "class State(TypedDict):\n",
    "    topic: str\n",
    "    joke: str\n",
    "\n",
    "def refine_topic(state: State):\n",
    "    return {\"topic\": state[\"topic\"] + \" and cats\"}\n",
    "\n",
    "def generate_joke(state: State):\n",
    "    return {\"joke\": f\"This is a joke about {state['topic']}\"}\n",
    "\n",
    "graph = (\n",
    "    StateGraph(State)\n",
    "    .add_node(refine_topic)\n",
    "    .add_node(generate_joke)\n",
    "    .add_edge(START, \"refine_topic\")\n",
    "    .add_edge(\"refine_topic\", \"generate_joke\")\n",
    "    .compile()\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a9abgx0zao5",
   "metadata": {},
   "source": [
    "#### 4.1.1.1 流式处理模式：`\"values\"`\n",
    "\n",
    "`\"values\"` 流式处理模式通过在每个步骤后流式传输图的完整状态，提供对图执行的高级视图。本质上，在 LangGraph 中的每个节点或超级步骤执行完毕后，整个状态对象（包含所有已定义的状态变量及其当前值）都会作为单个块在流中发出。\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "p72l0l15o39",
   "metadata": {},
   "source": [
    "##### 示例 4-2：使用 `stream_mode=\"values\"` 执行图\n",
    "\n",
    "当我们使用 `stream_mode=\"values\"` 执行此图时，我们会观察到在每个节点执行后状态都被流式传输：\"\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "5amzi1td7mx",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{'topic': 'ice cream'}\n",
      "{'topic': 'ice cream and cats'}\n",
      "{'topic': 'ice cream and cats', 'joke': 'This is a joke about ice cream and cats'}\n"
     ]
    }
   ],
   "source": [
    "for chunk in graph.stream(\n",
    "    {\"topic\": \"ice cream\"},\n",
    "    stream_mode=\"values\",\n",
    "):\n",
    "    print(chunk)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "xssnyt9znjb",
   "metadata": {},
   "source": [
    "**💡 核心概念解析**：\n",
    "\n",
    "正如您从输出中看到的，第一个块显示了仅包含 \"topic\" 的初始状态。第二个块反映了 `refine_topic` 节点执行后的状态，更新了 \"topic\"。最后，最后一个块显示了 `generate_joke` 节点执行后的完整状态，现在同时包含 \"topic\" 和 \"joke\"。这种逐步查看完整状态的方式是 `\"values\"` 流式处理的本质。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "rxeqhi9jlsf",
   "metadata": {},
   "source": [
    "#### 4.1.1.2 流式处理模式：`\"updates\"`\n",
    "\n",
    "与 `\"values\"` 相比，`\"updates\"` 流式处理模式提供了更集中和简洁的信息流。`\"updates\"` 不是流式传输整个状态，而是仅流式传输每个节点对状态进行的特定更新。这意味着您会收到键值对流，其中键是生成更新的节点的名称，值是一个字典，其中包含该节点修改的状态变量。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "riywzvs34kt",
   "metadata": {},
   "source": [
    "##### 示例 4-3：使用 `stream_mode=\"updates\"` 执行图\n",
    "\n",
    "使用与之前相同的笑话生成程序，让我们观察使用 `stream_mode=\"updates\"` 的输出："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "vqnohbbo8b8",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{'refine_topic': {'topic': 'ice cream and cats'}}\n",
      "{'generate_joke': {'joke': 'This is a joke about ice cream and cats'}}\n"
     ]
    }
   ],
   "source": [
    "for chunk in graph.stream(\n",
    "    {\"topic\": \"ice cream\"},\n",
    "    stream_mode=\"updates\",\n",
    "):\n",
    "    print(chunk)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "vw3ehxc73m",
   "metadata": {},
   "source": [
    "**💡 关键理解**：\n",
    "\n",
    "在这里，输出更加精简。我们看到与每个节点名称关联的更新。`'refine_topic'` 更新显示了 `topic` 的更改，`'generate_joke'` 更新显示了新生成的 `joke`。此模式有效地隔离并仅呈现每个步骤中发生的更改，使其成为有针对性的、逐步 UI 更新的理想选择。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "rnwoinhkbwq",
   "metadata": {},
   "source": [
    "#### 4.1.1.3 流式处理模式：`\"custom\"`\n",
    "\n",
    "`\"custom\"` 流式处理模式提供了极致的灵活性，允许开发人员从其 LangGraph 节点内部流式传输任意数据。这是通过 `StreamWriter` 对象实现的，该对象可以在任何节点函数中访问。节点可以使用 `StreamWriter` 在其执行期间的任何时间点将任何类型的数据发送到流中。\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "gv9b7hm2si",
   "metadata": {},
   "source": [
    "##### 示例 4-4：修改 `generate_joke` 节点以使用 `StreamWriter`\n",
    "\n",
    "让我们修改我们的 `generate_joke` 节点，使其使用 `StreamWriter` 在返回笑话之前发送一条自定义消息：\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "y05l5d4uyyr",
   "metadata": {},
   "outputs": [],
   "source": [
    "from langgraph.types import StreamWriter\n",
    "\n",
    "def generate_joke_with_custom(state: State, writer: StreamWriter):\n",
    "    writer({\"custom_key\": \"Writing custom data while generating a joke\"})\n",
    "    return {\"joke\": f\"This is a joke about {state['topic']}\"}\n",
    "\n",
    "graph_custom = (\n",
    "    StateGraph(State)\n",
    "    .add_node(refine_topic)\n",
    "    .add_node(\"generate_joke\", generate_joke_with_custom)\n",
    "    .add_edge(START, \"refine_topic\")\n",
    "    .add_edge(\"refine_topic\", \"generate_joke\")\n",
    "    .compile()\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "n5708j667ah",
   "metadata": {},
   "source": [
    "##### 示例 4-5：使用 `stream_mode=\"custom\"` 执行图\n",
    "\n",
    "现在，使用 `stream_mode=\"custom\"` 运行会产生以下结果：\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "v24rba5v9y",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{'custom_key': 'Writing custom data while generating a joke'}\n"
     ]
    }
   ],
   "source": [
    "for chunk in graph_custom.stream(\n",
    "    {\"topic\": \"ice cream\"},\n",
    "    stream_mode=\"custom\",\n",
    "):\n",
    "    print(chunk)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0o5neb985mxm",
   "metadata": {},
   "source": [
    "**💡 自定义流式处理的威力**：\n",
    "\n",
    "如示例所示，我们收到了通过 `StreamWriter` 发送的自定义字典。此模式允许将任何相关信息嵌入到流中，使其高度适应特定的应用程序需求。结合工具使用时，`\"custom\"` 流式处理在提供对复杂智能体行为的实时洞察方面变得更加强大。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "s4oxttjhf6",
   "metadata": {},
   "source": [
    "#### 4.1.1.4 流式处理模式：`\"messages\"`\n",
    "\n",
    "对于涉及 LLM 交互的应用程序，尤其是对话式智能体，`\"messages\"` 流式处理模式非常宝贵。此模式流式传输 LangGraph 中 LLM 生成的单个令牌，以及相关的元数据。这种令牌级流式处理允许实现最细粒度的实时反馈，从而在聊天机器人和其他文本生成应用程序中实现\"打字效果\"界面。\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "wtq85u84ijm",
   "metadata": {},
   "source": [
    "##### 示例 4-6：使用 LLM 生成笑话并启用 `stream_mode=\"messages\"`\n",
    "\n",
    "让我们增强我们的笑话生成图，使用 LLM 生成笑话，并观察 `\"messages\"` 流：\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "a0qex4xeee",
   "metadata": {},
   "outputs": [],
   "source": [
    "from langchain_openai import ChatOpenAI\n",
    "\n",
    "llm = ChatOpenAI(model=\"Qwen/Qwen2.5-7B-Instruct\")\n",
    "\n",
    "def generate_joke_with_llm(state: State):\n",
    "    llm_response = llm.invoke(\n",
    "        [{\"role\": \"user\", \"content\": f\"Generate a joke about {state['topic']}\"}]\n",
    "    )\n",
    "    return {\"joke\": llm_response.content}\n",
    "\n",
    "graph_with_llm = (\n",
    "    StateGraph(State)\n",
    "    .add_node(refine_topic)\n",
    "    .add_node(\"generate_joke\", generate_joke_with_llm)\n",
    "    .add_edge(START, \"refine_topic\")\n",
    "    .add_edge(\"refine_topic\", \"generate_joke\")\n",
    "    .compile()\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "w1f7h1k3mb",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Why| did| the| cat| sit| on| the| ice| cream| cone|?\n",
      "\n",
      "|Because| he| wanted| to| see| if| it| would| turn| into| a| laser| pointer|-fl|avored| dessert|!|"
     ]
    }
   ],
   "source": [
    "for message_chunk, metadata in graph_with_llm.stream(\n",
    "    {\"topic\": \"ice cream\"},\n",
    "    stream_mode=\"messages\",\n",
    "):\n",
    "    if message_chunk.content:\n",
    "        print(message_chunk.content, end=\"|\", flush=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ccw8ddvmx4s",
   "metadata": {},
   "source": [
    "**💡 词元级流式处理的魅力**：\n",
    "\n",
    "正如您所看到的，我们现在收到了一系列单个词元，从而可以逐词元甚至逐字符地显示正在生成的笑话。随附的 `metadata` 字典提供了关于每个词元的宝贵上下文。这种细粒度的流式处理是构建真正交互式聊天应用程序的关键技术。\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9ed8fe43",
   "metadata": {},
   "source": [
    "#### 4.1.1.5 流式处理模式：`\"debug\"`\n",
    "\n",
    "调试流模式专为需要全面监控 LangGraph 执行过程的开发者设计。该模式会流\n",
    "式传输包含丰富调试信息的事件流，在图执行的每个步骤提供详细数据，涵盖任务调\n",
    "度、执行结果、错误信息及状态转换等关键事件。调试流模式虽然因其技术性和信息\n",
    "密度较高而不适合直接面向用户展示，但在开发调试阶段和需要深度监控的场景中为\n",
    "开发者提供了不可或缺的分析工具，特别有助于进行性能优化和复杂行为分析。\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e1e3455b",
   "metadata": {},
   "source": [
    "##### 示例 4-7：使用 stream_mode=\"debug\" 执行图的输出"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "113a5b31",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{'step': 1, 'timestamp': '2025-08-20T13:50:34.979202+00:00', 'type': 'task', 'payload': {'id': 'bf38cfd7-04b6-a0c8-38d5-797dfc4b0771', 'name': 'refine_topic', 'input': {'topic': 'ice cream'}, 'triggers': ('branch:to:refine_topic',)}}\n",
      "{'step': 1, 'timestamp': '2025-08-20T13:50:34.980498+00:00', 'type': 'task_result', 'payload': {'id': 'bf38cfd7-04b6-a0c8-38d5-797dfc4b0771', 'name': 'refine_topic', 'error': None, 'result': [('topic', 'ice cream and cats')], 'interrupts': []}}\n",
      "{'step': 2, 'timestamp': '2025-08-20T13:50:34.980693+00:00', 'type': 'task', 'payload': {'id': 'af02992f-8f49-42e6-2689-9d685d0be885', 'name': 'generate_joke', 'input': {'topic': 'ice cream and cats'}, 'triggers': ('branch:to:generate_joke',)}}\n",
      "{'step': 2, 'timestamp': '2025-08-20T13:50:34.981762+00:00', 'type': 'task_result', 'payload': {'id': 'af02992f-8f49-42e6-2689-9d685d0be885', 'name': 'generate_joke', 'error': None, 'result': [('joke', 'This is a joke about ice cream and cats')], 'interrupts': []}}\n"
     ]
    }
   ],
   "source": [
    "for chunk in graph_custom.stream(\n",
    "    {\"topic\": \"ice cream\"},\n",
    "    stream_mode=\"debug\",\n",
    "):\n",
    "    print(chunk)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4a8c932d",
   "metadata": {},
   "source": [
    "#### 4.1.1.7 组合流式处理模式\n",
    "\n",
    "LangGraph 支持同时组合多种流式处理模式，通过将流模式字符串列表传递给\n",
    " `stream_mode` 参数，开发者可以接收包含多种模式数据的交错流。组合模式时，流式处理输出将变为 `(stream_mode, data)` 元组，其中 `stream_mode` 指示数据类型，data 是该模式的实际流式内容。此功能支持更丰富的监控和用户反馈场景，例如同时查看高级状态更新和细粒度的 LLM 词元流。\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3a469643",
   "metadata": {},
   "source": [
    "##### 示例 4-8：组合流式处理模式的输出"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "03ed76bc",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Stream mode: updates\n",
      "{'refine_topic': {'topic': 'ice cream and cats'}}\n",
      "\n",
      "\n",
      "Stream mode: custom\n",
      "{'custom_key': 'Writing custom data wHITLe generating a joke'}\n",
      "\n",
      "\n",
      "Stream mode: updates\n",
      "{'generate_joke': {'joke': 'This is a joke about ice cream and cats'}}\n",
      "\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from langgraph.types import StreamWriter\n",
    "\n",
    "def generate_joke(state: State, writer: StreamWriter):\n",
    "    writer({\"custom_key\": \"Writing custom data wHITLe generating a joke\"})\n",
    "    return {\"joke\": f\"This is a joke about {state['topic']}\"}\n",
    "\n",
    "graph = (\n",
    "    StateGraph(State)\n",
    "    .add_node(refine_topic)\n",
    "    .add_node(generate_joke)\n",
    "    .add_edge(START, \"refine_topic\")\n",
    "    .add_edge(\"refine_topic\", \"generate_joke\")\n",
    "    .compile()\n",
    ")\n",
    "\n",
    "for stream_mode, chunk in graph.stream(\n",
    "    {\"topic\": \"ice cream\"},\n",
    "    stream_mode=[\"updates\", \"custom\"],\n",
    "):\n",
    "    print(f\"Stream mode: {stream_mode}\")\n",
    "    print(chunk)\n",
    "    print(\"\\n\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6aeef285",
   "metadata": {},
   "source": [
    "### 4.1.2 事件流式处理\n",
    "\n",
    "除了 `stream()` 和 `astream()` 提供的基于模式的流式处理，LangGraph 还提供了 `astream_events()` 方法，用于访问图中执行期间发生的较低级别事件流。此方法适用于捕获节点内部事件，便于开发者更精细地了解图的内部工作原理。\n",
    "\n",
    "`astream_events()` 与 LangChain 对象中标准的事件流式处理接口一致，使其符合 LangChain 生态系统开发者的使用习惯。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5b79cb41",
   "metadata": {},
   "source": [
    "##### 示例 4-9：使用 astream_events 方法"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "28f63639",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "on_chain_start: LangGraph\n",
      "on_chain_start: call_model\n",
      "on_chat_model_start: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_stream: ChatOpenAI\n",
      "on_chat_model_end: ChatOpenAI\n",
      "on_chain_stream: call_model\n",
      "on_chain_end: call_model\n",
      "on_chain_stream: LangGraph\n",
      "on_chain_end: LangGraph\n"
     ]
    }
   ],
   "source": [
    "from langchain_openai import ChatOpenAI\n",
    "from langgraph.graph import StateGraph, MessagesState, START, END\n",
    "\n",
    "model = ChatOpenAI(model=\"Qwen/Qwen3-8B\")\n",
    "\n",
    "def call_model(state: MessagesState):\n",
    "    response = model.invoke(state['messages'])\n",
    "    return {\"messages\": response}\n",
    "\n",
    "workflow = StateGraph(MessagesState)\n",
    "workflow.add_node(call_model)\n",
    "workflow.add_edge(START, \"call_model\")\n",
    "workflow.add_edge(\"call_model\", END)\n",
    "app = workflow.compile()\n",
    "\n",
    "inputs = [{\"role\": \"user\", \"content\": \"hi!\"}]\n",
    "async for event in app.astream_events({\"messages\": inputs}, version=\"v1\"):\n",
    "    kind = event[\"event\"]\n",
    "    print(f\"{kind}: {event['name']}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "mmpfx3bjxrk",
   "metadata": {},
   "source": [
    "## 4.2 确保连续性：LangGraph 中的持久化\n",
    "\n",
    "在上一节中，我们探讨了流式处理作为一种增强 AI 智能体实时响应性的机制。然而，一个真正强大且以用户为中心的人工智能智能体需要的不仅仅是速度；还需要连续性。用户期望智能体能够记住过去的交互，跨多个会话保持上下文，并能优雅地从中断或错误中恢复。\n",
    "\n",
    "LangGraph 中的持久化，通过其内置的存档点系统实现，使智能体能够在交互过程中保持其状态，从而实现一系列高级功能：\n",
    "- **对话记忆**：跨会话保持对话历史\n",
    "- **断点续传**：从中断点恢复执行\n",
    "- **时间旅行调试**：回溯到历史状态进行调试\n",
    "- **状态分支**：从历史状态创建新的执行路径\n",
    "\n",
    "### 4.2.1 持久化的核心概念：线程和存档点\n",
    "\n",
    "在 LangGraph 中，**线程**代表图的独特执行上下文。每个线程都由唯一的 `thread_id` 标识。**存档点**是图在特定时间点的状态快照。当启用持久化时，LangGraph 会在图执行的每个超步后自动创建这些快照。\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "dhobctgmi6g",
   "metadata": {},
   "source": [
    "### 4.2.2 存档点器实现：选择您的持久性后端\n",
    "\n",
    "LangGraph 提供了几种内置的存档点器实现，每种实现都利用不同的存储后端。\n",
    "\n",
    "#### MemorySaver：内存存档点\n",
    "\n",
    "- **优点**：快速、易于设置\n",
    "- **缺点**：数据不持久，应用重启后丢失\n",
    "- **用途**：开发、测试、原型设计\n",
    "\n",
    "#### SqliteSaver：基于文件的持久化\n",
    "\n",
    "- **优点**：轻量级、跨应用重启持久\n",
    "- **缺点**：不适合高并发场景\n",
    "- **用途**：本地应用、小规模部署\n",
    "\n",
    "#### PostgresSaver：生产级持久化\n",
    "\n",
    "- **优点**：可靠、可扩展、支持并发\n",
    "- **缺点**：需要数据库服务器\n",
    "- **用途**：生产环境、多用户应用\n",
    "\n",
    "让我们通过实际例子来演示持久化的使用：\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6h4p0hrc4ft",
   "metadata": {},
   "source": [
    "##### 示例 4-10 & 4-11：使用 MemorySaver 实现持久化的笑话生成器\n",
    "\n",
    "让我们创建一个带有持久化功能的笑话生成器，演示如何保存和恢复状态："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "6dv850lkmkn",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "第一次运行：\n",
      "{'refine_topic': {'topic': 'ice cream and cats'}}\n",
      "{'generate_joke': {'joke': 'Why did the cat wear a scarf around its ice cream cone?\\n\\nBecause it was a chilly day and the cat wanted to keep its \"paws\" cool!'}}\n",
      "\n",
      "获取最终状态：\n",
      "{'topic': 'ice cream and cats', 'joke': 'Why did the cat wear a scarf around its ice cream cone?\\n\\nBecause it was a chilly day and the cat wanted to keep its \"paws\" cool!'}\n"
     ]
    }
   ],
   "source": [
    "from langgraph.checkpoint.memory import MemorySaver\n",
    "\n",
    "# 使用 LLM 版本的笑话生成器，但添加持久化\n",
    "graph_persistent = (\n",
    "    StateGraph(State)\n",
    "    .add_node(refine_topic)\n",
    "    .add_node(\"generate_joke\", generate_joke_with_llm)\n",
    "    .add_edge(START, \"refine_topic\")\n",
    "    .add_edge(\"refine_topic\", \"generate_joke\")\n",
    "    .compile(checkpointer=MemorySaver())  # 启用持久化\n",
    ")\n",
    "\n",
    "# 定义线程配置\n",
    "config = {\"configurable\": {\"thread_id\": \"my_thread_1\"}}\n",
    "\n",
    "print(\"第一次运行：\")\n",
    "for chunk in graph_persistent.stream(\n",
    "    {\"topic\": \"ice cream\"},\n",
    "    config=config,\n",
    "    stream_mode=\"updates\",\n",
    "):\n",
    "    print(chunk)\n",
    "\n",
    "print(\"\\n获取最终状态：\")\n",
    "print(graph_persistent.get_state(config).values)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "525a0f61",
   "metadata": {},
   "source": [
    "##### 示例 4-12：使用 SqliteSaver 编译具有持久化的 LangGraph"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "215608aa",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "第一次运行：\n",
      "{'refine_topic': {'topic': 'ice cream and cats'}}\n",
      "{'generate_joke': {'joke': 'Why did the cat refuse to eat the ice cream？\\n\\nBecause it was purr-fectly chilled!'}}\n",
      "\n",
      "获取最终状态：\n",
      "{'topic': 'ice cream and cats', 'joke': 'Why did the cat refuse to eat the ice cream？\\n\\nBecause it was purr-fectly chilled!'}\n"
     ]
    }
   ],
   "source": [
    "import sqlite3\n",
    "from langgraph.checkpoint.sqlite import SqliteSaver\n",
    "\n",
    "conn = sqlite3.connect(\"checkpoints.sqlite\", check_same_thread=False)\n",
    "checkpointer = SqliteSaver(conn)\n",
    "\n",
    "# 使用 LLM 版本的笑话生成器，但添加持久化\n",
    "graph_persistent = (\n",
    "    StateGraph(State)\n",
    "    .add_node(refine_topic)\n",
    "    .add_node(\"generate_joke\", generate_joke_with_llm)\n",
    "    .add_edge(START, \"refine_topic\")\n",
    "    .add_edge(\"refine_topic\", \"generate_joke\")\n",
    "    .compile(checkpointer=checkpointer)  # 启用持久化\n",
    ")\n",
    "\n",
    "# 定义线程配置\n",
    "config = {\"configurable\": {\"thread_id\": \"my_thread_1\"}}\n",
    "\n",
    "print(\"第一次运行：\")\n",
    "for chunk in graph_persistent.stream(\n",
    "    {\"topic\": \"ice cream\"},\n",
    "    config=config,\n",
    "    stream_mode=\"updates\",\n",
    "):\n",
    "    print(chunk)\n",
    "\n",
    "print(\"\\n获取最终状态：\")\n",
    "print(graph_persistent.get_state(config).values)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "xujpn9l6hk8",
   "metadata": {},
   "source": [
    "##### 示例 4-15：使用 graph.get_state_history 浏览执行历史记录\n",
    "\n",
    "持久化的一个强大功能是能够查看完整的状态历史："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "5ev4wggaxa",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "状态历史回顾：\n",
      "\n",
      "存档点 1:\n",
      "  ID: 1f07dcca...\n",
      "  步骤: {'source': 'loop', 'step': 2, 'parents': {}}\n",
      "  状态: {'topic': 'ice cream and cats', 'joke': 'Why did the cat refuse to eat the ice cream？\\n\\nBecause it was purr-fectly chilled!'}\n",
      "  下一个节点: ()\n",
      "--------------------------------------------------\n",
      "\n",
      "存档点 2:\n",
      "  ID: 1f07dcca...\n",
      "  步骤: {'source': 'loop', 'step': 1, 'parents': {}}\n",
      "  状态: {'topic': 'ice cream and cats'}\n",
      "  下一个节点: ('generate_joke',)\n",
      "--------------------------------------------------\n",
      "\n",
      "存档点 3:\n",
      "  ID: 1f07dcca...\n",
      "  步骤: {'source': 'loop', 'step': 0, 'parents': {}}\n",
      "  状态: {'topic': 'ice cream'}\n",
      "  下一个节点: ('refine_topic',)\n",
      "--------------------------------------------------\n",
      "\n",
      "存档点 4:\n",
      "  ID: 1f07dcca...\n",
      "  步骤: {'source': 'input', 'step': -1, 'parents': {}}\n",
      "  状态: {}\n",
      "  下一个节点: ('__start__',)\n",
      "--------------------------------------------------\n"
     ]
    }
   ],
   "source": [
    "print(\"状态历史回顾：\")\n",
    "state_history = list(graph_persistent.get_state_history(config))\n",
    "for i, snapshot in enumerate(state_history):\n",
    "    print(f\"\\n存档点 {i + 1}:\")\n",
    "    print(f\"  ID: {snapshot.config['configurable']['checkpoint_id'][:8]}...\")\n",
    "    print(f\"  步骤: {snapshot.metadata}\")\n",
    "    print(f\"  状态: {snapshot.values}\")\n",
    "    print(f\"  下一个节点: {snapshot.next}\")\n",
    "    print(\"-\" * 50)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4jc25pz882o",
   "metadata": {},
   "source": [
    "**💡 状态历史的价值**：\n",
    "\n",
    "通过查看状态历史，我们可以看到：\n",
    "- 每个存档点的唯一 ID\n",
    "- 执行的步骤和元数据信息\n",
    "- 每个阶段的完整状态\n",
    "- 计划执行的下一个节点\n",
    "\n",
    "这对于调试、监控和理解智能体的行为非常有价值。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "31e508b6",
   "metadata": {},
   "source": [
    "##### 示例 4-16：使用 graph.stream(..., checkpoint_id=...) 重放执行"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "04bba284",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "重放检查点：{'topic': 'ice cream and cats'}\n",
      "下一个节点：('generate_joke',)\n",
      "\n",
      "从检查点继续执行：\n",
      "{'topic': 'ice cream and cats'}\n",
      "{'topic': 'ice cream and cats', 'joke': 'Why did the cat join the ice cream shop\\'s loyalty program?\\n\\nBecause she heard they had a \"paw-some\" reward for paws-singly good reviewers!'}\n"
     ]
    }
   ],
   "source": [
    "# 正确的重放方式：从检查点继续执行\n",
    "# 选择一个中间存档点（有下一个步骤的）\n",
    "checkpoint_to_replay = state_history[1]  # 第2个存档点，状态为刚完成refine_topic\n",
    "print(f\"重放检查点：{checkpoint_to_replay.values}\")\n",
    "print(f\"下一个节点：{checkpoint_to_replay.next}\")\n",
    "\n",
    "# 从该检查点继续执行\n",
    "replay_config = checkpoint_to_replay.config\n",
    "print(\"\\n从检查点继续执行：\")\n",
    "for event in graph_persistent.stream(None, replay_config, stream_mode=\"values\"):\n",
    "    print(event)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "24381069",
   "metadata": {},
   "source": [
    "##### 示例 4-17：使用 graph.update_state(..., values) 从存档点分支执行"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "1c3c6d10",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{'topic': 'ice cream and dogs'}\n",
      "{'topic': 'ice cream and dogs', 'joke': 'Why did the dog refuse the ice cream?\\n\\nBecause he heard it was \"paw-sicly\" good!'}\n"
     ]
    }
   ],
   "source": [
    "checkpoint_to_branch = state_history[1] # 让我们从历史记录中的第 2 个存档点分支\n",
    "branch_config = checkpoint_to_branch.config\n",
    "\n",
    "# 创建一个新存档点，使用相同的 checkpoint_id 但更新状态\n",
    "new_branch = graph_persistent.update_state(\n",
    "    branch_config,\n",
    "    {\"topic\": \"ice cream and dogs\"}\n",
    ")\n",
    "\n",
    "# 恢复图的执行 - 它现在将使用分支的存档点\n",
    "for event in graph_persistent.stream(None, new_branch, stream_mode=\"values\"):\n",
    "    print(event)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "17ca9ba0",
   "metadata": {},
   "source": [
    "##### 示例 4-18：获取子图状态快照\n",
    "\n",
    "让我们创建一个包含子图的复杂智能体系统，演示层次化状态管理："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "4d2fe22b",
   "metadata": {},
   "outputs": [],
   "source": [
    "from typing import TypedDict\n",
    "from langgraph.graph import StateGraph, START, END\n",
    "from langgraph.checkpoint.memory import MemorySaver\n",
    "\n",
    "# 定义子图状态\n",
    "class SubtaskState(TypedDict):\n",
    "    subtask_input: str\n",
    "    subtask_result: str\n",
    "\n",
    "# 定义主图状态  \n",
    "class MainState(TypedDict):\n",
    "    main_topic: str\n",
    "    processed_data: str\n",
    "    final_output: str\n",
    "\n",
    "# 子图节点函数\n",
    "def subtask_processor(state: SubtaskState):\n",
    "    \"\"\"子图中的处理节点\"\"\"\n",
    "    return {\n",
    "        \"subtask_result\": f\"处理完成: {state['subtask_input']} -> 子任务结果\"\n",
    "    }\n",
    "\n",
    "def subtask_formatter(state: SubtaskState):\n",
    "    \"\"\"子图中的格式化节点\"\"\"\n",
    "    return {\n",
    "        \"subtask_result\": f\"[格式化] {state['subtask_result']}\"\n",
    "    }\n",
    "\n",
    "# 构建独立的子图（带持久化）\n",
    "subgraph = StateGraph(SubtaskState)\n",
    "subgraph.add_node(\"process\", subtask_processor)\n",
    "subgraph.add_node(\"format\", subtask_formatter)\n",
    "subgraph.add_edge(START, \"process\")\n",
    "subgraph.add_edge(\"process\", \"format\")\n",
    "subgraph.add_edge(\"format\", END)\n",
    "compiled_subgraph = subgraph.compile(checkpointer=MemorySaver())\n",
    "\n",
    "# 主图节点函数\n",
    "def prepare_data(state: MainState):\n",
    "    \"\"\"准备数据\"\"\"\n",
    "    return {\n",
    "        \"processed_data\": f\"预处理: {state['main_topic']}\"\n",
    "    }\n",
    "\n",
    "def call_subgraph_node(state: MainState):\n",
    "    \"\"\"调用子图的节点\"\"\"\n",
    "    # 使用独立的子图配置\n",
    "    subgraph_config = {\"configurable\": {\"thread_id\": f\"sub_{state['main_topic']}\"}}\n",
    "    \n",
    "    # 调用子图处理\n",
    "    subgraph_input = {\"subtask_input\": state[\"processed_data\"]}\n",
    "    subgraph_result = compiled_subgraph.invoke(subgraph_input, subgraph_config)\n",
    "    \n",
    "    return {\n",
    "        \"processed_data\": subgraph_result[\"subtask_result\"]\n",
    "    }\n",
    "\n",
    "def finalize_output(state: MainState):\n",
    "    \"\"\"生成最终输出\"\"\"\n",
    "    return {\n",
    "        \"final_output\": f\"最终结果: {state['processed_data']}\"\n",
    "    }\n",
    "\n",
    "# 构建主图\n",
    "main_graph = StateGraph(MainState)\n",
    "main_graph.add_node(\"prepare\", prepare_data)\n",
    "main_graph.add_node(\"subgraph_call\", call_subgraph_node)\n",
    "main_graph.add_node(\"finalize\", finalize_output)\n",
    "\n",
    "main_graph.add_edge(START, \"prepare\")\n",
    "main_graph.add_edge(\"prepare\", \"subgraph_call\")\n",
    "main_graph.add_edge(\"subgraph_call\", \"finalize\")\n",
    "main_graph.add_edge(\"finalize\", END)\n",
    "\n",
    "# 编译主图（带持久化）\n",
    "hierarchical_graph = main_graph.compile(checkpointer=MemorySaver())"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0b8c4051",
   "metadata": {},
   "source": [
    "现在让我们运行层次化智能体并演示如何获取包含子图的完整状态快照："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "57db982c",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "=== 运行层次化智能体 ===\n",
      "执行结果： {'main_topic': '数据分析任务', 'processed_data': '[格式化] 处理完成: 预处理: 数据分析任务 -> 子任务结果', 'final_output': '最终结果: [格式化] 处理完成: 预处理: 数据分析任务 -> 子任务结果'}\n",
      "\n",
      "==================================================\n",
      "=== 获取子图状态快照 ===\n",
      "Grandparent State (主图状态):\n",
      "{'main_topic': '数据分析任务', 'processed_data': '[格式化] 处理完成: 预处理: 数据分析任务 -> 子任务结果', 'final_output': '最终结果: [格式化] 处理完成: 预处理: 数据分析任务 -> 子任务结果'}\n",
      "---------------\n",
      "Parent Graph State (子图状态):\n",
      "{'subtask_input': '预处理: 数据分析任务', 'subtask_result': '[格式化] 处理完成: 预处理: 数据分析任务 -> 子任务结果'}\n",
      "---------------\n",
      "Subgraph State History (子图历史状态):\n",
      "  子图快照 1: {'subtask_input': '预处理: 数据分析任务', 'subtask_result': '[格式化] 处理完成: 预处理: 数据分析任务 -> 子任务结果'}\n",
      "  子图快照 2: {'subtask_input': '预处理: 数据分析任务', 'subtask_result': '处理完成: 预处理: 数据分析任务 -> 子任务结果'}\n",
      "  子图快照 3: {'subtask_input': '预处理: 数据分析任务'}\n",
      "\n",
      "=== 当前示例的状态访问 ===\n",
      "主图最终状态： {'main_topic': '数据分析任务', 'processed_data': '[格式化] 处理完成: 预处理: 数据分析任务 -> 子任务结果', 'final_output': '最终结果: [格式化] 处理完成: 预处理: 数据分析任务 -> 子任务结果'}\n",
      "子图最终状态： {'subtask_input': '预处理: 数据分析任务', 'subtask_result': '[格式化] 处理完成: 预处理: 数据分析任务 -> 子任务结果'}\n",
      "✅ 通过独立的状态管理，我们可以分别访问主图和子图的完整执行历史\n"
     ]
    }
   ],
   "source": [
    "# 运行层次化智能体\n",
    "config = {\"configurable\": {\"thread_id\": \"hierarchical_thread\"}}\n",
    "\n",
    "print(\"=== 运行层次化智能体 ===\")\n",
    "result = hierarchical_graph.invoke(\n",
    "    {\"main_topic\": \"数据分析任务\"}, \n",
    "    config=config\n",
    ")\n",
    "print(\"执行结果：\", result)\n",
    "\n",
    "print(\"\\n\" + \"=\"*50)\n",
    "print(\"=== 获取子图状态快照 ===\")\n",
    "\n",
    "# 获取主图状态\n",
    "main_state = hierarchical_graph.get_state(config, subgraphs=True)\n",
    "print(\"Grandparent State (主图状态):\")\n",
    "print(main_state.values)  # 打印主图的状态值\n",
    "print(\"---------------\")\n",
    "\n",
    "# 获取子图状态（子图有独立的thread_id）\n",
    "subgraph_config = {\"configurable\": {\"thread_id\": \"sub_数据分析任务\"}}\n",
    "subgraph_state = compiled_subgraph.get_state(subgraph_config, subgraphs=True)\n",
    "print(\"Parent Graph State (子图状态):\")\n",
    "print(subgraph_state.values)  # 打印子图的状态值\n",
    "print(\"---------------\")\n",
    "\n",
    "# 检查子图的历史状态\n",
    "print(\"Subgraph State History (子图历史状态):\")\n",
    "sub_history = list(compiled_subgraph.get_state_history(subgraph_config))\n",
    "for i, snapshot in enumerate(sub_history[:3]):\n",
    "    print(f\"  子图快照 {i+1}: {snapshot.values}\")\n",
    "\n",
    "print(\"\\n=== 当前示例的状态访问 ===\")\n",
    "print(\"主图最终状态：\", main_state.values)\n",
    "print(\"子图最终状态：\", subgraph_state.values)\n",
    "print(\"✅ 通过独立的状态管理，我们可以分别访问主图和子图的完整执行历史\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "pq60sga94ze",
   "metadata": {},
   "source": [
    "##### 示例 4-19：更新子图状态\n",
    "\n",
    "让我们演示如何更新主图和子图的状态，并观察状态变化："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "jxdht2prsh",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 首先，让我们创建一个包含更多状态字段的增强版子图\n",
    "from typing import TypedDict\n",
    "\n",
    "class EnhancedSubtaskState(TypedDict):\n",
    "    subtask_input: str\n",
    "    subtask_result: str\n",
    "    city: str  # 新增城市字段\n",
    "    temperature: int  # 新增温度字段\n",
    "\n",
    "class EnhancedMainState(TypedDict):\n",
    "    main_topic: str\n",
    "    processed_data: str\n",
    "    final_output: str\n",
    "    user_location: str  # 新增用户位置字段\n",
    "\n",
    "# 增强的子图节点函数\n",
    "def enhanced_subtask_processor(state: EnhancedSubtaskState):\n",
    "    \"\"\"增强的子图处理节点\"\"\"\n",
    "    city = state.get(\"city\", \"未知城市\")\n",
    "    temp = state.get(\"temperature\", 20)\n",
    "    return {\n",
    "        \"subtask_result\": f\"处理完成: {state['subtask_input']} (位置: {city}, 温度: {temp}°C) -> 子任务结果\"\n",
    "    }\n",
    "\n",
    "def enhanced_subtask_formatter(state: EnhancedSubtaskState):\n",
    "    \"\"\"增强的子图格式化节点\"\"\"\n",
    "    return {\n",
    "        \"subtask_result\": f\"[格式化] {state['subtask_result']}\"\n",
    "    }\n",
    "\n",
    "# 构建增强版子图\n",
    "enhanced_subgraph = StateGraph(EnhancedSubtaskState)\n",
    "enhanced_subgraph.add_node(\"process\", enhanced_subtask_processor)\n",
    "enhanced_subgraph.add_node(\"format\", enhanced_subtask_formatter)\n",
    "enhanced_subgraph.add_edge(START, \"process\")\n",
    "enhanced_subgraph.add_edge(\"process\", \"format\")\n",
    "enhanced_subgraph.add_edge(\"format\", END)\n",
    "compiled_enhanced_subgraph = enhanced_subgraph.compile(checkpointer=MemorySaver())\n",
    "\n",
    "# 增强版主图节点函数\n",
    "def enhanced_prepare_data(state: EnhancedMainState):\n",
    "    \"\"\"增强的数据准备\"\"\"\n",
    "    return {\n",
    "        \"processed_data\": f\"预处理: {state['main_topic']}\",\n",
    "        \"user_location\": state.get(\"user_location\", \"默认位置\")\n",
    "    }\n",
    "\n",
    "def enhanced_call_subgraph_node(state: EnhancedMainState):\n",
    "    \"\"\"调用增强版子图的节点\"\"\"\n",
    "    subgraph_config = {\"configurable\": {\"thread_id\": f\"enhanced_sub_{state['main_topic']}\"}}\n",
    "    \n",
    "    # 调用子图处理，传递更多参数\n",
    "    subgraph_input = {\n",
    "        \"subtask_input\": state[\"processed_data\"],\n",
    "        \"city\": \"beijing\",  # 默认城市\n",
    "        \"temperature\": 25   # 默认温度\n",
    "    }\n",
    "    subgraph_result = compiled_enhanced_subgraph.invoke(subgraph_input, subgraph_config)\n",
    "    \n",
    "    return {\n",
    "        \"processed_data\": subgraph_result[\"subtask_result\"]\n",
    "    }\n",
    "\n",
    "def enhanced_finalize_output(state: EnhancedMainState):\n",
    "    \"\"\"生成增强的最终输出\"\"\"\n",
    "    return {\n",
    "        \"final_output\": f\"最终结果: {state['processed_data']} (用户位置: {state.get('user_location', '未知')})\"\n",
    "    }\n",
    "\n",
    "# 构建增强版主图\n",
    "enhanced_main_graph = StateGraph(EnhancedMainState)\n",
    "enhanced_main_graph.add_node(\"prepare\", enhanced_prepare_data)\n",
    "enhanced_main_graph.add_node(\"subgraph_call\", enhanced_call_subgraph_node)\n",
    "enhanced_main_graph.add_node(\"finalize\", enhanced_finalize_output)\n",
    "\n",
    "enhanced_main_graph.add_edge(START, \"prepare\")\n",
    "enhanced_main_graph.add_edge(\"prepare\", \"subgraph_call\")\n",
    "enhanced_main_graph.add_edge(\"subgraph_call\", \"finalize\")\n",
    "enhanced_main_graph.add_edge(\"finalize\", END)\n",
    "\n",
    "# 编译增强版主图\n",
    "enhanced_hierarchical_graph = enhanced_main_graph.compile(checkpointer=MemorySaver())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "suptevtvcwr",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "=== 步骤 1: 运行增强版层次化智能体 ===\n",
      "初始执行结果： {'main_topic': '天气分析任务', 'processed_data': '[格式化] 处理完成: 预处理: 天气分析任务 (位置: beijing, 温度: 25°C) -> 子任务结果', 'final_output': '最终结果: [格式化] 处理完成: 预处理: 天气分析任务 (位置: beijing, 温度: 25°C) -> 子任务结果 (用户位置: shanghai)', 'user_location': 'shanghai'}\n",
      "\n",
      "============================================================\n",
      "=== 步骤 2: 获取当前状态 ===\n",
      "主图当前状态： {'main_topic': '天气分析任务', 'processed_data': '[格式化] 处理完成: 预处理: 天气分析任务 (位置: beijing, 温度: 25°C) -> 子任务结果', 'final_output': '最终结果: [格式化] 处理完成: 预处理: 天气分析任务 (位置: beijing, 温度: 25°C) -> 子任务结果 (用户位置: shanghai)', 'user_location': 'shanghai'}\n",
      "子图当前状态： {'subtask_input': '预处理: 天气分析任务', 'subtask_result': '[格式化] 处理完成: 预处理: 天气分析任务 (位置: beijing, 温度: 25°C) -> 子任务结果', 'city': 'beijing', 'temperature': 25}\n",
      "\n",
      "============================================================\n",
      "=== 步骤 3: 更新主图状态 ===\n",
      "更新主图状态: user_location 从 'shanghai' 改为 'guangzhou'\n",
      "主图状态更新完成，新配置： {'configurable': {'thread_id': 'enhanced_hierarchical_thread', 'checkpoint_ns': '', 'checkpoint_id': '1f07dccb-1910-6134-8004-0cb0146968a7'}}\n",
      "更新后的主图状态： {'main_topic': '天气分析任务', 'processed_data': '[格式化] 处理完成: 预处理: 天气分析任务 (位置: beijing, 温度: 25°C) -> 子任务结果', 'final_output': '最终结果: [格式化] 处理完成: 预处理: 天气分析任务 (位置: beijing, 温度: 25°C) -> 子任务结果 (用户位置: shanghai)', 'user_location': 'guangzhou'}\n",
      "\n",
      "============================================================\n",
      "=== 步骤 4: 更新子图状态 ===\n",
      "更新子图状态: city 从 'beijing' 改为 'la', temperature 从 25 改为 18\n",
      "子图状态更新完成，新配置： {'configurable': {'thread_id': 'enhanced_sub_天气分析任务', 'checkpoint_ns': '', 'checkpoint_id': '1f07dccb-1910-690e-8003-72bbf49d8012'}}\n",
      "更新后的子图状态： {'subtask_input': '预处理: 天气分析任务', 'subtask_result': '[格式化] 处理完成: 预处理: 天气分析任务 (位置: beijing, 温度: 25°C) -> 子任务结果', 'city': 'la', 'temperature': 18}\n",
      "\n",
      "============================================================\n",
      "=== 步骤 5: 从更新的状态继续执行 ===\n",
      "从更新的子图状态继续执行:\n",
      "子图执行结果： {'subtask_input': '预处理: 天气分析任务', 'subtask_result': '[格式化] 处理完成: 预处理: 天气分析任务 (位置: beijing, 温度: 25°C) -> 子任务结果', 'city': 'la', 'temperature': 18}\n"
     ]
    }
   ],
   "source": [
    "# 运行增强版智能体并演示状态更新\n",
    "print(\"=== 步骤 1: 运行增强版层次化智能体 ===\")\n",
    "enhanced_config = {\"configurable\": {\"thread_id\": \"enhanced_hierarchical_thread\"}}\n",
    "\n",
    "# 初始运行\n",
    "initial_result = enhanced_hierarchical_graph.invoke(\n",
    "    {\"main_topic\": \"天气分析任务\", \"user_location\": \"shanghai\"}, \n",
    "    enhanced_config\n",
    ")\n",
    "print(\"初始执行结果：\", initial_result)\n",
    "\n",
    "print(\"\\n\" + \"=\"*60)\n",
    "print(\"=== 步骤 2: 获取当前状态 ===\")\n",
    "\n",
    "# 获取主图状态\n",
    "main_state = enhanced_hierarchical_graph.get_state(enhanced_config)\n",
    "print(\"主图当前状态：\", main_state.values)\n",
    "\n",
    "# 获取子图状态\n",
    "subgraph_config = {\"configurable\": {\"thread_id\": \"enhanced_sub_天气分析任务\"}}\n",
    "subgraph_state = compiled_enhanced_subgraph.get_state(subgraph_config)\n",
    "print(\"子图当前状态：\", subgraph_state.values)\n",
    "\n",
    "print(\"\\n\" + \"=\"*60)\n",
    "print(\"=== 步骤 3: 更新主图状态 ===\")\n",
    "\n",
    "# 更新主图状态中的 user_location 字段\n",
    "print(\"更新主图状态: user_location 从 'shanghai' 改为 'guangzhou'\")\n",
    "updated_main_config = enhanced_hierarchical_graph.update_state(\n",
    "    enhanced_config,\n",
    "    {\"user_location\": \"guangzhou\"}\n",
    ")\n",
    "print(\"主图状态更新完成，新配置：\", updated_main_config)\n",
    "\n",
    "# 查看更新后的主图状态\n",
    "updated_main_state = enhanced_hierarchical_graph.get_state(updated_main_config)\n",
    "print(\"更新后的主图状态：\", updated_main_state.values)\n",
    "\n",
    "print(\"\\n\" + \"=\"*60)\n",
    "print(\"=== 步骤 4: 更新子图状态 ===\")\n",
    "\n",
    "# 更新子图状态中的 city 字段（按照你的要求）\n",
    "print(\"更新子图状态: city 从 'beijing' 改为 'la', temperature 从 25 改为 18\")\n",
    "updated_subgraph_config = compiled_enhanced_subgraph.update_state(\n",
    "    subgraph_config,  # 将子图状态的 config 作为第一个参数传入\n",
    "    {\"city\": \"la\", \"temperature\": 18}  # updates 参数指定要更新的状态键值对\n",
    ")\n",
    "print(\"子图状态更新完成，新配置：\", updated_subgraph_config)\n",
    "\n",
    "# 查看更新后的子图状态\n",
    "updated_subgraph_state = compiled_enhanced_subgraph.get_state(updated_subgraph_config)\n",
    "print(\"更新后的子图状态：\", updated_subgraph_state.values)\n",
    "\n",
    "print(\"\\n\" + \"=\"*60)\n",
    "print(\"=== 步骤 5: 从更新的状态继续执行 ===\")\n",
    "\n",
    "# 从更新的子图状态继续执行（重新处理）\n",
    "print(\"从更新的子图状态继续执行:\")\n",
    "for chunk in compiled_enhanced_subgraph.stream(None, updated_subgraph_config, stream_mode=\"values\"):\n",
    "    print(\"子图执行结果：\", chunk)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "k98hg03un",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "=== 步骤 6: 以子图节点身份更新状态 ===\n",
      "运行新的实例以获取子图执行状态...\n",
      "初始测试结果： {'main_topic': '测试任务', 'processed_data': '[格式化] 处理完成: 预处理: 测试任务 (位置: beijing, 温度: 25°C) -> 子任务结果', 'final_output': '最终结果: [格式化] 处理完成: 预处理: 测试任务 (位置: beijing, 温度: 25°C) -> 子任务结果 (用户位置: beijing)', 'user_location': 'beijing'}\n",
      "\n",
      "获取子图状态以进行节点身份更新...\n",
      "当前子图状态： {'subtask_input': '预处理: 测试任务', 'subtask_result': '[格式化] 处理完成: 预处理: 测试任务 (位置: beijing, 温度: 25°C) -> 子任务结果', 'city': 'beijing', 'temperature': 25}\n",
      "\n",
      "=== 以 'process' 节点身份更新子图状态 ===\n",
      "以 'process' 节点身份更新状态完成！\n",
      "更新后的配置： {'configurable': {'thread_id': 'enhanced_sub_测试任务', 'checkpoint_ns': '', 'checkpoint_id': '1f07dccb-1924-63fa-8003-718cc2c95b4d'}}\n",
      "以节点身份更新后的子图状态： {'subtask_input': '预处理: 测试任务', 'subtask_result': '模拟process节点执行结果', 'city': 'la', 'temperature': 18}\n",
      "下一个要执行的节点： ('format',)\n",
      "\n",
      "=== 从节点身份更新的状态继续执行 ===\n",
      "从 'process' 节点更新的状态继续执行后续节点...\n",
      "继续执行结果： {'subtask_input': '预处理: 测试任务', 'subtask_result': '模拟process节点执行结果', 'city': 'la', 'temperature': 18}\n",
      "继续执行结果： {'subtask_input': '预处理: 测试任务', 'subtask_result': '[格式化] 模拟process节点执行结果', 'city': 'la', 'temperature': 18}\n",
      "\n",
      "💡 以节点身份更新状态的关键概念：\n",
      "1. as_node='process' 参数模拟了 process 节点执行后的状态更新效果\n",
      "2. 这种方式实现了更精细的子图状态控制，允许精确模拟子图内部节点行为\n",
      "3. 更新后可以继续执行后续节点，支持高级流程干预和定制需求\n",
      "4. 特别适用于测试、调试和模拟特定节点执行结果的场景\n"
     ]
    }
   ],
   "source": [
    "# 演示以子图节点身份更新状态\n",
    "print(\"=== 步骤 6: 以子图节点身份更新状态 ===\")\n",
    "\n",
    "# 首先运行一个新的实例来获取活跃的子图状态\n",
    "print(\"运行新的实例以获取子图执行状态...\")\n",
    "test_config = {\"configurable\": {\"thread_id\": \"test_node_update\"}}\n",
    "test_subgraph_config = {\"configurable\": {\"thread_id\": \"enhanced_sub_测试任务\"}}\n",
    "\n",
    "# 先运行主图\n",
    "test_result = enhanced_hierarchical_graph.invoke(\n",
    "    {\"main_topic\": \"测试任务\", \"user_location\": \"beijing\"}, \n",
    "    test_config\n",
    ")\n",
    "\n",
    "print(\"初始测试结果：\", test_result)\n",
    "\n",
    "print(\"\\n获取子图状态以进行节点身份更新...\")\n",
    "current_subgraph_state = compiled_enhanced_subgraph.get_state(test_subgraph_config)\n",
    "print(\"当前子图状态：\", current_subgraph_state.values)\n",
    "\n",
    "print(\"\\n=== 以 'process' 节点身份更新子图状态 ===\")\n",
    "# 以子图 enhanced_subgraph 的 process 节点身份更新状态\n",
    "updated_config_as_node = compiled_enhanced_subgraph.update_state(\n",
    "    test_subgraph_config,  # 将子图状态的 config 作为第一个参数传入\n",
    "    {\"city\": \"la\", \"temperature\": 18, \"subtask_result\": \"模拟process节点执行结果\"},  # updates 参数指定\"虚假的\"处理结果\n",
    "    as_node=\"process\",  # 指定要模拟的子图节点名称为 process\n",
    ")\n",
    "\n",
    "print(\"以 'process' 节点身份更新状态完成！\")\n",
    "print(\"更新后的配置：\", updated_config_as_node)\n",
    "\n",
    "# 查看更新后的状态\n",
    "updated_state_as_node = compiled_enhanced_subgraph.get_state(updated_config_as_node)\n",
    "print(\"以节点身份更新后的子图状态：\", updated_state_as_node.values)\n",
    "print(\"下一个要执行的节点：\", updated_state_as_node.next)\n",
    "\n",
    "print(\"\\n=== 从节点身份更新的状态继续执行 ===\")\n",
    "print(\"从 'process' 节点更新的状态继续执行后续节点...\")\n",
    "\n",
    "# 从更新的状态继续执行（会执行format节点）\n",
    "for chunk in compiled_enhanced_subgraph.stream(None, updated_config_as_node, stream_mode=\"values\"):\n",
    "    print(\"继续执行结果：\", chunk)\n",
    "\n",
    "print(\"\\n💡 以节点身份更新状态的关键概念：\")\n",
    "print(\"1. as_node='process' 参数模拟了 process 节点执行后的状态更新效果\")\n",
    "print(\"2. 这种方式实现了更精细的子图状态控制，允许精确模拟子图内部节点行为\")\n",
    "print(\"3. 更新后可以继续执行后续节点，支持高级流程干预和定制需求\")\n",
    "print(\"4. 特别适用于测试、调试和模拟特定节点执行结果的场景\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9t03x2evrid",
   "metadata": {},
   "source": [
    "##### 示例 4-20：以子图节点身份更新状态\n",
    "\n",
    "演示如何以特定子图节点的身份更新状态，实现更精细的状态控制："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "vqe8o0ll0y",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 创建一个更真实的天气查询子图示例\n",
    "from typing import TypedDict, List\n",
    "from langchain_core.messages import BaseMessage, HumanMessage, AIMessage\n",
    "\n",
    "class WeatherState(TypedDict):\n",
    "    location: str\n",
    "    messages: List[BaseMessage]\n",
    "    weather_result: str\n",
    "\n",
    "class ParentTaskState(TypedDict):\n",
    "    task_name: str\n",
    "    location: str\n",
    "    weather_info: str\n",
    "    final_report: str\n",
    "\n",
    "# 天气子图节点函数\n",
    "def weather_analyzer(state: WeatherState):\n",
    "    \"\"\"分析天气请求\"\"\"\n",
    "    location = state.get(\"location\", \"unknown\")\n",
    "    return {\n",
    "        \"messages\": state[\"messages\"] + [HumanMessage(content=f\"请查询{location}的天气\")],\n",
    "        \"location\": location\n",
    "    }\n",
    "\n",
    "def weather_node(state: WeatherState):\n",
    "    \"\"\"天气查询节点 - 这是我们要模拟的关键节点\"\"\"\n",
    "    location = state.get(\"location\", \"unknown\")\n",
    "    # 模拟天气API调用\n",
    "    weather_result = f\"{location}的天气是晴朗的\"\n",
    "    return {\n",
    "        \"messages\": state[\"messages\"] + [AIMessage(content=weather_result)],\n",
    "        \"weather_result\": weather_result\n",
    "    }\n",
    "\n",
    "def weather_formatter(state: WeatherState):\n",
    "    \"\"\"格式化天气结果\"\"\"\n",
    "    weather_result = state.get(\"weather_result\", \"无天气数据\")\n",
    "    return {\n",
    "        \"weather_result\": f\"[格式化] {weather_result}\"\n",
    "    }\n",
    "\n",
    "# 构建天气子图 (weather_graph)\n",
    "weather_graph = StateGraph(WeatherState)\n",
    "weather_graph.add_node(\"weather_analyzer\", weather_analyzer)\n",
    "weather_graph.add_node(\"weather_node\", weather_node)  # 关键的weather_node\n",
    "weather_graph.add_node(\"weather_formatter\", weather_formatter)\n",
    "\n",
    "weather_graph.add_edge(START, \"weather_analyzer\")\n",
    "weather_graph.add_edge(\"weather_analyzer\", \"weather_node\")\n",
    "weather_graph.add_edge(\"weather_node\", \"weather_formatter\")\n",
    "weather_graph.add_edge(\"weather_formatter\", END)\n",
    "\n",
    "compiled_weather_graph = weather_graph.compile(checkpointer=MemorySaver())\n",
    "\n",
    "# 父图节点函数\n",
    "def prepare_weather_task(state: ParentTaskState):\n",
    "    \"\"\"准备天气查询任务\"\"\"\n",
    "    return {\n",
    "        \"task_name\": f\"天气查询任务: {state.get('task_name', '默认任务')}\",\n",
    "        \"location\": state.get(\"location\", \"beijing\")\n",
    "    }\n",
    "\n",
    "def call_weather_subgraph(state: ParentTaskState):\n",
    "    \"\"\"调用天气子图\"\"\"\n",
    "    subgraph_config = {\"configurable\": {\"thread_id\": f\"weather_{state['location']}\"}}\n",
    "    \n",
    "    # 调用天气子图\n",
    "    subgraph_input = {\n",
    "        \"location\": state[\"location\"],\n",
    "        \"messages\": [HumanMessage(content=\"开始天气查询\")],\n",
    "        \"weather_result\": \"\"\n",
    "    }\n",
    "    subgraph_result = compiled_weather_graph.invoke(subgraph_input, subgraph_config)\n",
    "    \n",
    "    return {\n",
    "        \"weather_info\": subgraph_result[\"weather_result\"]\n",
    "    }\n",
    "\n",
    "def generate_final_report(state: ParentTaskState):\n",
    "    \"\"\"生成最终报告\"\"\"\n",
    "    return {\n",
    "        \"final_report\": f\"任务完成: {state['task_name']}, 天气信息: {state.get('weather_info', '无')}\"\n",
    "    }\n",
    "\n",
    "# 构建父图\n",
    "parent_weather_graph = StateGraph(ParentTaskState)\n",
    "parent_weather_graph.add_node(\"prepare_task\", prepare_weather_task)\n",
    "parent_weather_graph.add_node(\"call_weather\", call_weather_subgraph)\n",
    "parent_weather_graph.add_node(\"generate_report\", generate_final_report)\n",
    "\n",
    "parent_weather_graph.add_edge(START, \"prepare_task\")\n",
    "parent_weather_graph.add_edge(\"prepare_task\", \"call_weather\")\n",
    "parent_weather_graph.add_edge(\"call_weather\", \"generate_report\")\n",
    "parent_weather_graph.add_edge(\"generate_report\", END)\n",
    "\n",
    "compiled_parent_weather_graph = parent_weather_graph.compile(checkpointer=MemorySaver())"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0a71c4a2",
   "metadata": {},
   "source": [
    "##### 示例 4-20：以子图节点身份更新状态"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "726e29a2",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "运行新的实例以获取子图执行状态...\n",
      "初始测试结果： {'main_topic': '测试任务', 'processed_data': '[格式化] 处理完成: 预处理: 测试任务 (位置: beijing, 温度: 25°C) -> 子任务结果', 'final_output': '最终结果: [格式化] 处理完成: 预处理: 测试任务 (位置: beijing, 温度: 25°C) -> 子任务结果 (用户位置: beijing)', 'user_location': 'beijing'}\n",
      "\n",
      "获取子图状态以进行节点身份更新...\n",
      "当前子图状态： {'subtask_input': '预处理: 测试任务', 'subtask_result': '[格式化] 处理完成: 预处理: 测试任务 (位置: beijing, 温度: 25°C) -> 子任务结果', 'city': 'beijing', 'temperature': 25}\n",
      "\n",
      "=== 以 'process' 节点身份更新子图状态 ===\n",
      "以 'process' 节点身份更新状态完成！\n",
      "更新后的配置： {'configurable': {'thread_id': 'enhanced_sub_测试任务', 'checkpoint_ns': '', 'checkpoint_id': '1f07dccb-1953-6fd8-8009-840394da26e1'}}\n",
      "以节点身份更新后的子图状态： {'subtask_input': '预处理: 测试任务', 'subtask_result': '模拟process节点执行结果', 'city': 'la', 'temperature': 18}\n",
      "下一个要执行的节点： ('format',)\n",
      "\n",
      "=== 从节点身份更新的状态继续执行 ===\n",
      "从 'process' 节点更新的状态继续执行后续节点...\n",
      "继续执行结果： {'subtask_input': '预处理: 测试任务', 'subtask_result': '模拟process节点执行结果', 'city': 'la', 'temperature': 18}\n",
      "继续执行结果： {'subtask_input': '预处理: 测试任务', 'subtask_result': '[格式化] 模拟process节点执行结果', 'city': 'la', 'temperature': 18}\n"
     ]
    }
   ],
   "source": [
    "# 首先运行一个新的实例来获取活跃的子图状态\n",
    "print(\"运行新的实例以获取子图执行状态...\")\n",
    "test_config = {\"configurable\": {\"thread_id\": \"test_node_update\"}}\n",
    "test_subgraph_config = {\"configurable\": {\"thread_id\":\n",
    "\"enhanced_sub_测试任务\"}}\n",
    "\n",
    "# 先运行主图\n",
    "test_result = enhanced_hierarchical_graph.invoke(\n",
    "    {\"main_topic\": \"测试任务\", \"user_location\": \"beijing\"},\n",
    "    test_config\n",
    ")\n",
    "\n",
    "print(\"初始测试结果：\", test_result)\n",
    "\n",
    "print(\"\\n获取子图状态以进行节点身份更新...\")\n",
    "current_subgraph_state = compiled_enhanced_subgraph.get_state(test_subgraph_config)\n",
    "print(\"当前子图状态：\", current_subgraph_state.values)\n",
    "\n",
    "print(\"\\n=== 以 'process' 节点身份更新子图状态 ===\")\n",
    "# 以子图 enhanced_subgraph 的 process 节点身份更新状态\n",
    "updated_config_as_node = compiled_enhanced_subgraph.update_state(\n",
    "    test_subgraph_config,  # 将子图状态的 config 作为第一个参数传入\n",
    "    {\"city\": \"la\", \"temperature\": 18, \"subtask_result\":\n",
    "\"模拟process节点执行结果\"},  # updates 参数指定\"虚假的\"处理结果\n",
    "    as_node=\"process\",  # 指定要模拟的子图节点名称为 process\n",
    ")\n",
    "\n",
    "print(\"以 'process' 节点身份更新状态完成！\")\n",
    "print(\"更新后的配置：\", updated_config_as_node)\n",
    "\n",
    "# 查看更新后的状态\n",
    "updated_state_as_node = compiled_enhanced_subgraph.get_state(updated_config_as_node)\n",
    "print(\"以节点身份更新后的子图状态：\", updated_state_as_node.values)\n",
    "print(\"下一个要执行的节点：\", updated_state_as_node.next)\n",
    "\n",
    "print(\"\\n=== 从节点身份更新的状态继续执行 ===\")\n",
    "print(\"从 'process' 节点更新的状态继续执行后续节点...\")\n",
    "\n",
    "# 从更新的状态继续执行（会执行format节点）\n",
    "for chunk in compiled_enhanced_subgraph.stream(None,\n",
    "updated_config_as_node, stream_mode=\"values\"):\n",
    "    print(\"继续执行结果：\", chunk)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "44jlwpqvowm",
   "metadata": {},
   "source": [
    "## 4.3 人机协作：构建可控的智能体系统\n",
    "\n",
    "虽然流式处理和持久化大大增强了智能体的用户体验和可靠性，但在许多实际应用中，我们还需要一个关键组件：**人机环路协作（Human-in-the-loop）**。这种机制允许人类用户在智能体执行的关键节点进行干预、指导或批准，确保智能体的行为符合人类的期望和价值观。\n",
    "\n",
    "人机环路协作特别重要的场景包括：\n",
    "- **高风险决策**：金融交易、医疗诊断、法律建议等\n",
    "- **创意协作**：内容创作、设计决策、策略规划等\n",
    "- **质量控制**：确保输出符合特定标准和要求\n",
    "- **学习改进**：通过人类反馈改善智能体行为\n",
    "\n",
    "### 4.3.1 中断机制：智能体的\"暂停键\"\n",
    "\n",
    "LangGraph 提供了两种主要的中断机制：\n",
    "\n",
    "1. **静态中断**：在编译时定义的固定中断点\n",
    "   - `interrupt_before=\"node_name\"`：在指定节点执行前中断\n",
    "   - `interrupt_after=\"node_name\"`：在指定节点执行后中断\n",
    "\n",
    "2. **动态中断**：通过 `interrupt()` 函数在节点内部触发的中断\n",
    "   - 更灵活，可以基于运行时条件决定是否中断\n",
    "   - 可以传递数据给人类用户进行决策\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a4lej3asxot",
   "metadata": {},
   "source": [
    "##### 示例 4-22：使用 interrupt() 函数的“人工审批节点”以及基于人工输入（批准 / 拒绝）的路由逻辑\n",
    "\n",
    "让我们构建一个智能体，它会在执行关键操作前请求人工审批："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "htghvpdimib",
   "metadata": {},
   "outputs": [],
   "source": [
    "from typing import Literal\n",
    "from langgraph.types import interrupt, Command\n",
    "\n",
    "# 定义状态类型\n",
    "class ApprovalState(TypedDict):\n",
    "    topic: str\n",
    "    proposed_action_details: str\n",
    "    final_result: str\n",
    "\n",
    "# 定义节点函数\n",
    "def propose_action(state: ApprovalState) -> ApprovalState:\n",
    "    \"\"\"提出一个需要人类审批的操作\"\"\"\n",
    "    return {\n",
    "        **state,\n",
    "        \"proposed_action_details\": f\"基于主题 '{state['topic']}' 的操作提议：发送营销邮件给1000个客户\"\n",
    "    }\n",
    "\n",
    "def human_approval_node(state: ApprovalState) -> Command[Literal[\"execute_action\", \"revise_action\"]]:\n",
    "    \"\"\"获取人类审批的节点\"\"\"\n",
    "    approval_request = interrupt(\n",
    "        {\n",
    "            \"question\": \"是否批准执行以下操作？\",\n",
    "            \"action_details\": state[\"proposed_action_details\"],\n",
    "            \"options\": [\"approve\", \"deny\"]\n",
    "        }\n",
    "    )\n",
    "\n",
    "    if approval_request.get(\"user_response\") == \"approve\":\n",
    "        return Command(goto=\"execute_action\")\n",
    "    else:\n",
    "        return Command(goto=\"revise_action\")\n",
    "\n",
    "def execute_action(state: ApprovalState) -> ApprovalState:\n",
    "    \"\"\"执行已批准的操作\"\"\"\n",
    "    return {\n",
    "        **state,\n",
    "        \"final_result\": f\"✅ 已执行操作: {state['proposed_action_details']}\"\n",
    "    }\n",
    "\n",
    "def revise_action(state: ApprovalState) -> ApprovalState:\n",
    "    \"\"\"修改被拒绝的操作\"\"\"\n",
    "    return {\n",
    "        **state, \n",
    "        \"final_result\": f\"❌ 操作被拒绝，已修改为: 发送营销邮件给50个目标客户（缩小规模）\"\n",
    "    }"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "r967n77llj",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 构建需要人工审批的图\n",
    "approval_graph_builder = StateGraph(ApprovalState)\n",
    "\n",
    "# 添加节点\n",
    "approval_graph_builder.add_node(\"propose_action\", propose_action)\n",
    "approval_graph_builder.add_node(\"human_approval\", human_approval_node)\n",
    "approval_graph_builder.add_node(\"execute_action\", execute_action)\n",
    "approval_graph_builder.add_node(\"revise_action\", revise_action)\n",
    "\n",
    "# 添加边\n",
    "approval_graph_builder.add_edge(START, \"propose_action\")\n",
    "approval_graph_builder.add_edge(\"propose_action\", \"human_approval\")\n",
    "approval_graph_builder.add_edge(\"revise_action\", \"human_approval\")  # 修改后再次请求审批\n",
    "\n",
    "# 编译图（必须包含检查点器以支持中断）\n",
    "approval_graph = approval_graph_builder.compile(checkpointer=MemorySaver())"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ru348eqx2c",
   "metadata": {},
   "source": [
    "现在让我们测试这个人工审批智能体的工作流程：\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "7meefxj997",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "=== 第一步：启动智能体，等待人工审批 ===\n",
      "执行完成： {'topic': '产品推广活动', 'proposed_action_details': \"基于主题 '产品推广活动' 的操作提议：发送营销邮件给1000个客户\", '__interrupt__': [Interrupt(value={'question': '是否批准执行以下操作？', 'action_details': \"基于主题 '产品推广活动' 的操作提议：发送营销邮件给1000个客户\", 'options': ['approve', 'deny']}, id='bcb607546bc01bba5ec6be4582d8ee95')]}\n",
      "\n",
      "当前状态: {'topic': '产品推广活动', 'proposed_action_details': \"基于主题 '产品推广活动' 的操作提议：发送营销邮件给1000个客户\"}\n",
      "下一个节点: ('human_approval',)\n",
      "是否被中断: PregelTask(id='1cb3f54d-2f12-5706-a34d-88c0f13a1bda', name='human_approval', path=('__pregel_pull', 'human_approval'), error=None, interrupts=(Interrupt(value={'question': '是否批准执行以下操作？', 'action_details': \"基于主题 '产品推广活动' 的操作提议：发送营销邮件给1000个客户\", 'options': ['approve', 'deny']}, id='bcb607546bc01bba5ec6be4582d8ee95'),), state=None, result=None)\n"
     ]
    }
   ],
   "source": [
    "# 启动审批流程\n",
    "config = {\"configurable\": {\"thread_id\": \"approval_thread\"}}\n",
    "\n",
    "print(\"=== 第一步：启动智能体，等待人工审批 ===\")\n",
    "try:\n",
    "    result = approval_graph.invoke(\n",
    "        {\"topic\": \"产品推广活动\"}, \n",
    "        config=config\n",
    "    )\n",
    "    print(\"执行完成：\", result)\n",
    "except Exception as e:\n",
    "    print(f\"智能体在等待人工审批处中断: {e}\")\n",
    "    \n",
    "# 检查当前状态\n",
    "current_state = approval_graph.get_state(config)\n",
    "print(f\"\\n当前状态: {current_state.values}\")\n",
    "print(f\"下一个节点: {current_state.next}\")\n",
    "print(f\"是否被中断: {current_state.tasks[0] if current_state.tasks else 'No interrupts'}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "c8wrzch92vj",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "=== 第二步：人工拒绝审批 ===\n",
      "拒绝后的执行结果： {'topic': '产品推广活动', 'proposed_action_details': \"基于主题 '产品推广活动' 的操作提议：发送营销邮件给1000个客户\", 'final_result': '❌ 操作被拒绝，已修改为: 发送营销邮件给50个目标客户（缩小规模）', '__interrupt__': [Interrupt(value={'question': '是否批准执行以下操作？', 'action_details': \"基于主题 '产品推广活动' 的操作提议：发送营销邮件给1000个客户\", 'options': ['approve', 'deny']}, id='d190af3ae0a9a6d9e79508ad44ee627b')]}\n",
      "\n",
      "更新后状态: {'topic': '产品推广活动', 'proposed_action_details': \"基于主题 '产品推广活动' 的操作提议：发送营销邮件给1000个客户\", 'final_result': '❌ 操作被拒绝，已修改为: 发送营销邮件给50个目标客户（缩小规模）'}\n",
      "下一个节点: ('human_approval',)\n"
     ]
    }
   ],
   "source": [
    "# 模拟人工拒绝审批\n",
    "print(\"\\n=== 第二步：人工拒绝审批 ===\")\n",
    "try:\n",
    "    result = approval_graph.invoke(\n",
    "        Command(resume={\"user_response\": \"deny\"}), \n",
    "        config=config\n",
    "    )\n",
    "    print(\"拒绝后的执行结果：\", result)\n",
    "except Exception as e:\n",
    "    print(f\"继续等待审批: {e}\")\n",
    "    \n",
    "# 再次检查状态\n",
    "current_state = approval_graph.get_state(config)\n",
    "print(f\"\\n更新后状态: {current_state.values}\")\n",
    "print(f\"下一个节点: {current_state.next}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "id": "1144zbce357r",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "=== 第三步：批准修改后的操作 ===\n",
      "最终执行结果： ✅ 已执行操作: 基于主题 '产品推广活动' 的操作提议：发送营销邮件给1000个客户\n"
     ]
    }
   ],
   "source": [
    "# 最终批准修改后的操作\n",
    "print(\"\\n=== 第三步：批准修改后的操作 ===\")\n",
    "final_result = approval_graph.invoke(\n",
    "    Command(resume={\"user_response\": \"approve\"}), \n",
    "    config=config\n",
    ")\n",
    "print(\"最终执行结果：\", final_result[\"final_result\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "id": "1f7864d3",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "StateSnapshot(values={'topic': '产品推广活动', 'proposed_action_details': \"基于主题 '产品推广活动' 的操作提议：发送营销邮件给1000个客户\", 'final_result': \"✅ 已执行操作: 基于主题 '产品推广活动' 的操作提议：发送营销邮件给1000个客户\"}, next=(), config={'configurable': {'thread_id': 'approval_thread', 'checkpoint_ns': '', 'checkpoint_id': '1f07dccb-1994-6b0a-8005-9e3b0eb5ea80'}}, metadata={'source': 'loop', 'step': 5, 'parents': {}}, created_at='2025-08-20T13:50:55.489918+00:00', parent_config={'configurable': {'thread_id': 'approval_thread', 'checkpoint_ns': '', 'checkpoint_id': '1f07dccb-1993-6d86-8004-4ac9c814bad6'}}, tasks=(), interrupts=())\n"
     ]
    }
   ],
   "source": [
    "print(approval_graph.get_state(config))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ebe6e05f",
   "metadata": {},
   "source": [
    "##### 示例 4-23：使用 interrupt() 函数的“人工编辑节点”以及使用人工编辑的值更新图状态"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "id": "a68fd92a",
   "metadata": {},
   "outputs": [],
   "source": [
    "from typing import Annotated\n",
    "from langgraph.graph import StateGraph, START, END\n",
    "from langgraph.graph.message import add_messages\n",
    "from langgraph.types import interrupt, Command\n",
    "from langgraph.checkpoint.memory import MemorySaver\n",
    "from langchain_core.messages import HumanMessage, AIMessage\n",
    "\n",
    "# 定义完整的状态类\n",
    "class HumanInLoopState(TypedDict):\n",
    "    user_question: str\n",
    "    generated_summary: str\n",
    "    messages: Annotated[list, add_messages]\n",
    "\n",
    "def summary_generation_node(state: HumanInLoopState):\n",
    "    \"\"\"生成初始摘要的节点\"\"\"\n",
    "    user_question = state[\"user_question\"]\n",
    "\n",
    "    # 模拟LLM生成摘要\n",
    "    generated_summary = f\"\"\"Based on the question \"{user_question}\", \n",
    "here's an initial summary:\n",
    "    \n",
    "LangGraph is a framework for building AI agents with the following \n",
    "features:\n",
    "- State management capabilities\n",
    "- Graph-based workflow design\n",
    "- Integration with LangChain ecosystem\n",
    "- Support for complex agent interactions\n",
    "\n",
    "This summary needs human review for accuracy and completeness.\"\"\"\n",
    "\n",
    "    return {\n",
    "        \"generated_summary\": generated_summary,\n",
    "        \"messages\": [AIMessage(content=f\"Generated initial summary for: {user_question}\")]\n",
    "    }\n",
    "\n",
    "def human_review_node(state: HumanInLoopState):\n",
    "    \"\"\"人工审核节点 - 使用 Interrupt 请求人工干预\"\"\"\n",
    "    current_summary = state[\"generated_summary\"]\n",
    "\n",
    "    # 触发中断，请求人工审核\n",
    "    edited_summary = interrupt(\n",
    "        {\n",
    "            \"task\": \"Please review and edit the generated summary\",\n",
    "            \"current_summary\": current_summary,\n",
    "            \"instructions\": \"Improve accuracy, add missing details, and ensure clarity\"\n",
    "        }\n",
    "    )\n",
    "\n",
    "    return {\n",
    "        \"generated_summary\": edited_summary,\n",
    "        \"messages\": [AIMessage(content=\"Summary sent for human review\")]\n",
    "    }\n",
    "\n",
    "# 创建状态图\n",
    "graph_builder = StateGraph(HumanInLoopState)\n",
    "\n",
    "# 添加节点\n",
    "graph_builder.add_node(\"generate_summary\", summary_generation_node)\n",
    "graph_builder.add_node(\"human_review\", human_review_node)\n",
    "\n",
    "# 定义边\n",
    "graph_builder.add_edge(START, \"generate_summary\")\n",
    "graph_builder.add_edge(\"generate_summary\", \"human_review\")\n",
    "graph_builder.add_edge(\"human_review\", END)\n",
    "\n",
    "# 编译图 - 在人工审核节点前中断\n",
    "memory_saver = MemorySaver()\n",
    "complete_human_loop_graph = graph_builder.compile(\n",
    "    checkpointer=memory_saver,\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "id": "57ab408d",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "🚀 第一步：运行图直到人工审核节点...\n",
      "==================================================\n",
      "📊 第一步结果: {'user_question': 'What are the key advantages of LangGraph for enterprise AI applications?', 'generated_summary': 'Based on the question \"What are the key advantages of LangGraph for enterprise AI applications?\", \\nhere\\'s an initial summary:\\n\\nLangGraph is a framework for building AI agents with the following \\nfeatures:\\n- State management capabilities\\n- Graph-based workflow design\\n- Integration with LangChain ecosystem\\n- Support for complex agent interactions\\n\\nThis summary needs human review for accuracy and completeness.', 'messages': [HumanMessage(content='Please generate a comprehensive summary', additional_kwargs={}, response_metadata={}, id='3713a1a2-4871-4403-bf24-ca0faecc6c31'), AIMessage(content='Generated initial summary for: What are the key advantages of LangGraph for enterprise AI applications?', additional_kwargs={}, response_metadata={}, id='a2bde527-d564-46c2-bd0b-2ffadfa4d6d6')], '__interrupt__': [Interrupt(value={'task': 'Please review and edit the generated summary', 'current_summary': 'Based on the question \"What are the key advantages of LangGraph for enterprise AI applications?\", \\nhere\\'s an initial summary:\\n\\nLangGraph is a framework for building AI agents with the following \\nfeatures:\\n- State management capabilities\\n- Graph-based workflow design\\n- Integration with LangChain ecosystem\\n- Support for complex agent interactions\\n\\nThis summary needs human review for accuracy and completeness.', 'instructions': 'Improve accuracy, add missing details, and ensure clarity'}, id='68da2689c4711efc14e68284023749a5')]}\n",
      "\n",
      "🔍 当前状态值: {'user_question': 'What are the key advantages of LangGraph for enterprise AI applications?', 'generated_summary': 'Based on the question \"What are the key advantages of LangGraph for enterprise AI applications?\", \\nhere\\'s an initial summary:\\n\\nLangGraph is a framework for building AI agents with the following \\nfeatures:\\n- State management capabilities\\n- Graph-based workflow design\\n- Integration with LangChain ecosystem\\n- Support for complex agent interactions\\n\\nThis summary needs human review for accuracy and completeness.', 'messages': [HumanMessage(content='Please generate a comprehensive summary', additional_kwargs={}, response_metadata={}, id='3713a1a2-4871-4403-bf24-ca0faecc6c31'), AIMessage(content='Generated initial summary for: What are the key advantages of LangGraph for enterprise AI applications?', additional_kwargs={}, response_metadata={}, id='a2bde527-d564-46c2-bd0b-2ffadfa4d6d6')]}\n",
      "⏭️ 下一个待执行节点: ('human_review',)\n",
      "📋 待处理任务: (PregelTask(id='b405b0d8-b0c7-c6d8-6d20-37aee25317aa', name='human_review', path=('__pregel_pull', 'human_review'), error=None, interrupts=(Interrupt(value={'task': 'Please review and edit the generated summary', 'current_summary': 'Based on the question \"What are the key advantages of LangGraph for enterprise AI applications?\", \\nhere\\'s an initial summary:\\n\\nLangGraph is a framework for building AI agents with the following \\nfeatures:\\n- State management capabilities\\n- Graph-based workflow design\\n- Integration with LangChain ecosystem\\n- Support for complex agent interactions\\n\\nThis summary needs human review for accuracy and completeness.', 'instructions': 'Improve accuracy, add missing details, and ensure clarity'}, id='68da2689c4711efc14e68284023749a5'),), state=None, result=None),)\n",
      "🎯 任务详情: PregelTask(id='b405b0d8-b0c7-c6d8-6d20-37aee25317aa', name='human_review', path=('__pregel_pull', 'human_review'), error=None, interrupts=(Interrupt(value={'task': 'Please review and edit the generated summary', 'current_summary': 'Based on the question \"What are the key advantages of LangGraph for enterprise AI applications?\", \\nhere\\'s an initial summary:\\n\\nLangGraph is a framework for building AI agents with the following \\nfeatures:\\n- State management capabilities\\n- Graph-based workflow design\\n- Integration with LangChain ecosystem\\n- Support for complex agent interactions\\n\\nThis summary needs human review for accuracy and completeness.', 'instructions': 'Improve accuracy, add missing details, and ensure clarity'}, id='68da2689c4711efc14e68284023749a5'),), state=None, result=None)\n",
      "⚠️ 中断数据: Interrupt(value={'task': 'Please review and edit the generated summary', 'current_summary': 'Based on the question \"What are the key advantages of LangGraph for enterprise AI applications?\", \\nhere\\'s an initial summary:\\n\\nLangGraph is a framework for building AI agents with the following \\nfeatures:\\n- State management capabilities\\n- Graph-based workflow design\\n- Integration with LangChain ecosystem\\n- Support for complex agent interactions\\n\\nThis summary needs human review for accuracy and completeness.', 'instructions': 'Improve accuracy, add missing details, and ensure clarity'}, id='68da2689c4711efc14e68284023749a5')\n",
      "📝 需要审核的内容: {'task': 'Please review and edit the generated summary', 'current_summary': 'Based on the question \"What are the key advantages of LangGraph for enterprise AI applications?\", \\nhere\\'s an initial summary:\\n\\nLangGraph is a framework for building AI agents with the following \\nfeatures:\\n- State management capabilities\\n- Graph-based workflow design\\n- Integration with LangChain ecosystem\\n- Support for complex agent interactions\\n\\nThis summary needs human review for accuracy and completeness.', 'instructions': 'Improve accuracy, add missing details, and ensure clarity'}\n"
     ]
    }
   ],
   "source": [
    "# 💡 第一步：运行图直到中断点\n",
    "thread_config = {\"configurable\": {\"thread_id\": \"complete_human_loop\"}}\n",
    "\n",
    "print(\"🚀 第一步：运行图直到人工审核节点...\")\n",
    "print(\"=\" * 50)\n",
    "\n",
    "# 初始运行 - 会在 human_review 节点前停止\n",
    "first_result = complete_human_loop_graph.invoke(\n",
    "    {\n",
    "        \"user_question\": \"What are the key advantages of LangGraph for enterprise AI applications?\",\n",
    "        \"messages\": [HumanMessage(content=\"Please generate a comprehensive summary\")]\n",
    "    },\n",
    "    config=thread_config\n",
    ")\n",
    "\n",
    "print(f\"📊 第一步结果: {first_result}\")\n",
    "print()\n",
    "\n",
    "# 检查当前图状态\n",
    "current_state = complete_human_loop_graph.get_state(thread_config)\n",
    "print(f\"🔍 当前状态值: {current_state.values}\")\n",
    "print(f\"⏭️ 下一个待执行节点: {current_state.next}\")\n",
    "print(f\"📋 待处理任务: {current_state.tasks}\")\n",
    "\n",
    "if current_state.tasks:\n",
    "    task_info = current_state.tasks[0]\n",
    "    print(f\"🎯 任务详情: {task_info}\")\n",
    "    if hasattr(task_info, 'interrupts') and task_info.interrupts:\n",
    "        interrupt_data = task_info.interrupts[0]\n",
    "        print(f\"⚠️ 中断数据: {interrupt_data}\")\n",
    "        if hasattr(interrupt_data, 'value'):\n",
    "            print(f\"📝 需要审核的内容: {interrupt_data.value}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "id": "77f31b35",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "🚀 第二步：提供人工审核后的改进摘要...\n",
      "==================================================\n",
      "✏️ 人工改进的摘要:\n",
      "LangGraph provides significant advantages for \n",
      "enterprise AI applications:\n",
      "\n",
      "🏗️ **Enterprise-Grade Architecture**:\n",
      "- Robust state management with built-in persistence\n",
      "- Scalable graph-based workflow orchestration\n",
      "- Production-ready error handling and recovery\n",
      "\n",
      "🔧 **Advanced Integration Capabilities**:\n",
      "- Seamless integration with LangChain ecosystem\n",
      "- Support for multiple LLM providers and tools\n",
      "- Human-in-the-loop workflows for quality assurance\n",
      "\n",
      "⚡ **Performance & User Experience**:\n",
      "- Real-time streaming capabilities for responsive interfaces\n",
      "- Efficient checkpoint and resume functionality\n",
      "- Memory management for long-running conversations\n",
      "\n",
      "🛡️ **Enterprise Features**:\n",
      "- Comprehensive monitoring and observability\n",
      "- Security controls and access management\n",
      "- Deployment flexibility across cloud and on-premise\n",
      "\n",
      "This makes LangGraph an ideal choice for mission-critical enterprise AI \n",
      "agent deployments.\n",
      "\n",
      "📏 摘要长度: 907 字符\n"
     ]
    }
   ],
   "source": [
    "# 💡 第二步：模拟人工审核并提供改进的摘要\n",
    "print(\"\\n🚀 第二步：提供人工审核后的改进摘要...\")\n",
    "print(\"=\" * 50)\n",
    "\n",
    "# 模拟人工审核后的改进摘要\n",
    "improved_summary = \"\"\"LangGraph provides significant advantages for \n",
    "enterprise AI applications:\n",
    "\n",
    "🏗️ **Enterprise-Grade Architecture**:\n",
    "- Robust state management with built-in persistence\n",
    "- Scalable graph-based workflow orchestration\n",
    "- Production-ready error handling and recovery\n",
    "\n",
    "🔧 **Advanced Integration Capabilities**:\n",
    "- Seamless integration with LangChain ecosystem\n",
    "- Support for multiple LLM providers and tools\n",
    "- Human-in-the-loop workflows for quality assurance\n",
    "\n",
    "⚡ **Performance & User Experience**:\n",
    "- Real-time streaming capabilities for responsive interfaces\n",
    "- Efficient checkpoint and resume functionality\n",
    "- Memory management for long-running conversations\n",
    "\n",
    "🛡️ **Enterprise Features**:\n",
    "- Comprehensive monitoring and observability\n",
    "- Security controls and access management\n",
    "- Deployment flexibility across cloud and on-premise\n",
    "\n",
    "This makes LangGraph an ideal choice for mission-critical enterprise AI \n",
    "agent deployments.\"\"\"\n",
    "\n",
    "print(f\"✏️ 人工改进的摘要:\\n{improved_summary}\")\n",
    "print(f\"\\n📏 摘要长度: {len(improved_summary)} 字符\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "id": "c2706d47",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "🚀 第三步：恢复图执行，使用人工审核的摘要...\n",
      "==================================================\n",
      "✅ 最终执行结果: {'user_question': 'What are the key advantages of LangGraph for enterprise AI applications?', 'generated_summary': 'LangGraph provides significant advantages for \\nenterprise AI applications:\\n\\n🏗️ **Enterprise-Grade Architecture**:\\n- Robust state management with built-in persistence\\n- Scalable graph-based workflow orchestration\\n- Production-ready error handling and recovery\\n\\n🔧 **Advanced Integration Capabilities**:\\n- Seamless integration with LangChain ecosystem\\n- Support for multiple LLM providers and tools\\n- Human-in-the-loop workflows for quality assurance\\n\\n⚡ **Performance & User Experience**:\\n- Real-time streaming capabilities for responsive interfaces\\n- Efficient checkpoint and resume functionality\\n- Memory management for long-running conversations\\n\\n🛡️ **Enterprise Features**:\\n- Comprehensive monitoring and observability\\n- Security controls and access management\\n- Deployment flexibility across cloud and on-premise\\n\\nThis makes LangGraph an ideal choice for mission-critical enterprise AI \\nagent deployments.', 'messages': [HumanMessage(content='Please generate a comprehensive summary', additional_kwargs={}, response_metadata={}, id='3713a1a2-4871-4403-bf24-ca0faecc6c31'), AIMessage(content='Generated initial summary for: What are the key advantages of LangGraph for enterprise AI applications?', additional_kwargs={}, response_metadata={}, id='a2bde527-d564-46c2-bd0b-2ffadfa4d6d6'), AIMessage(content='Summary sent for human review', additional_kwargs={}, response_metadata={}, id='d44f98e3-ccc2-4bf6-8d2d-eedb42ea31f4')]}\n",
      "\n",
      "🎯 最终状态: {'user_question': 'What are the key advantages of LangGraph for enterprise AI applications?', 'generated_summary': 'LangGraph provides significant advantages for \\nenterprise AI applications:\\n\\n🏗️ **Enterprise-Grade Architecture**:\\n- Robust state management with built-in persistence\\n- Scalable graph-based workflow orchestration\\n- Production-ready error handling and recovery\\n\\n🔧 **Advanced Integration Capabilities**:\\n- Seamless integration with LangChain ecosystem\\n- Support for multiple LLM providers and tools\\n- Human-in-the-loop workflows for quality assurance\\n\\n⚡ **Performance & User Experience**:\\n- Real-time streaming capabilities for responsive interfaces\\n- Efficient checkpoint and resume functionality\\n- Memory management for long-running conversations\\n\\n🛡️ **Enterprise Features**:\\n- Comprehensive monitoring and observability\\n- Security controls and access management\\n- Deployment flexibility across cloud and on-premise\\n\\nThis makes LangGraph an ideal choice for mission-critical enterprise AI \\nagent deployments.', 'messages': [HumanMessage(content='Please generate a comprehensive summary', additional_kwargs={}, response_metadata={}, id='3713a1a2-4871-4403-bf24-ca0faecc6c31'), AIMessage(content='Generated initial summary for: What are the key advantages of LangGraph for enterprise AI applications?', additional_kwargs={}, response_metadata={}, id='a2bde527-d564-46c2-bd0b-2ffadfa4d6d6'), AIMessage(content='Summary sent for human review', additional_kwargs={}, response_metadata={}, id='d44f98e3-ccc2-4bf6-8d2d-eedb42ea31f4')]}\n",
      "📝 最终摘要长度: 907 字符\n",
      "\n",
      "📨 消息历史:\n",
      "  1. Please generate a comprehensive summary\n",
      "  2. Generated initial summary for: What are the key advantages of LangGraph for enterprise AI applications?\n",
      "  3. Summary sent for human review\n",
      "\n",
      "🏁 人工在环流程完成！摘要已通过人工审核并改进。\n"
     ]
    }
   ],
   "source": [
    "# 💡 第三步：使用 Command 恢复图执行，传入人工审核的结果\n",
    "print(\"🚀 第三步：恢复图执行，使用人工审核的摘要...\")\n",
    "print(\"=\" * 50)\n",
    "\n",
    "# 使用 Command 恢复图执行，传入改进后的摘要\n",
    "final_result = complete_human_loop_graph.invoke(\n",
    "    Command(resume=improved_summary),  # 将改进的摘要作为中断节点的返回值\n",
    "    config=thread_config\n",
    ")\n",
    "\n",
    "print(f\"✅ 最终执行结果: {final_result}\")\n",
    "print()\n",
    "\n",
    "# 获取最终状态\n",
    "final_state = complete_human_loop_graph.get_state(thread_config)\n",
    "print(f\"🎯 最终状态: {final_state.values}\")\n",
    "print(f\"📝 最终摘要长度: {len(final_result.get('generated_summary', ''))} 字符\")\n",
    "print()\n",
    "\n",
    "# 显示消息历史\n",
    "print(\"📨 消息历史:\")\n",
    "for i, msg in enumerate(final_result.get('messages', []), 1):\n",
    "    print(f\"  {i}. {msg.content}\")\n",
    "\n",
    "print(f\"\\n🏁 人工在环流程完成！摘要已通过人工审核并改进。\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "wshuffvh1kq",
   "metadata": {},
   "source": [
    "**💡 人机环路协作的关键优势**：\n",
    "\n",
    "1. **质量控制**：人类可以在关键节点进行质量检查和决策\n",
    "2. **风险控制**：避免智能体执行高风险或不当操作\n",
    "3. **学习机会**：通过人类反馈改善智能体的决策能力\n",
    "4. **灵活性**：支持复杂的多轮交互和动态决策路径\n",
    "5. **可信度**：增强用户对智能体系统的信任和控制感\n",
    "\n",
    "这种人机协作模式特别适用于需要人类监督和指导的复杂智能体应用场景。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1jcpj9z4ky7h",
   "metadata": {},
   "source": "## 📚 本章总结\n\n通过本章的学习，我们深入掌握了 LangGraph 框架中提升 AI 智能体交互体验的三大核心技术。首先学习了流式处理技术，掌握了 `values`、`updates`、`custom`、`messages`、`debug` 等多种流式模式，能够克服 LLM 延迟并提供实时用户反馈。接着探讨了持久化机制，了解如何跨会话维护智能体记忆、实现时间旅行调试和断点续传功能。最后深入研究了人机环路协作，学会在关键决策点引入人类监督，通过静态和动态中断机制确保智能体行为的安全性和可控性。这些技术的结合使我们能够构建不仅智能，而且值得信赖、响应迅速、用户友好的 AI 智能体系统，为后续构建复杂的多智能体协作场景奠定了坚实基础。"
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": ".venv",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.13.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}